package com.yj.rocketmq.consumer;

import com.yj.core.exception.ServiceException;
import com.yj.rocketmq.MqConstants;
import com.yj.rocketmq.RocketMqConstants;
import com.yj.rocketmq.RocketMqExceptionMsg;
import com.yj.rocketmq.listener.AbstractMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import java.util.Arrays;
import java.util.List;

/**
 * Mq消息监听
 *
 * @author Xie
 * @date 19-3-19
 */
@Slf4j
public abstract class AbstractMsgConsumer {

    /**
     * mq消费者初始化
     *
     * @param orderMessageListener
     */
    public void run(AbstractMessageListener orderMessageListener) {

        //检查是否需要启用消费者
        if (getConsumerEnable() == null || !getConsumerEnable()) {
            log.info("{}未启用DefaultMQPushConsumer，无效初始化", RocketMqConstants.LOG_PREFIX);
            return;
        }
        //检查配置参数
        if (StringUtils.isAnyBlank(getNameSrv(), getConsumerGroup(), getTopic())) {
            log.error("请正确配置，MQ的Consumer参数");
            throw new ServiceException(RocketMqExceptionMsg.TOPIC_NOT_NULL);
        }

        log.info("{} DefaultMQPushConsumer参数:nameSrv={}，consumerGroup={}",
                RocketMqConstants.LOG_PREFIX, getNameSrv(), getConsumerGroup());

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getConsumerGroup());
        consumer.setNamesrvAddr(getNameSrv());
        try {
            String[] split = getTopic().split(",");
            List<String> topics = Arrays.asList(split);
            for (String topic : topics) {
                String tags = getTags(topic);

                log.info("{}订阅消息 topic={},tags={} ",
                        RocketMqConstants.LOG_PREFIX, topic.trim().toUpperCase(), tags.trim().toUpperCase());

                consumer.subscribe(topic.trim().toUpperCase(), tags.trim().toUpperCase());
            }

            // 从消息队列头部开始消费
            // !warn 新建订阅组的时候，消费进度是存在broker上的，获取topic对应的队列的最小偏移量，
            // 如果broker端的队列中的数据不多，还在内存中，那么偏移量将返回为0，即从头消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(orderMessageListener);
            // !warn 默认情况下不需要设置instanceName
            // rocketmq会使用ip@pid(pid代表jvm名字)作为唯一标示（如192.168.1.66＠26749），
            // 那么会导致多个consumer内部只有一个MQClientInstance(与mq交互)实例，从而导致只往一个集群发消息。
            consumer.setInstanceName(getConsumerGroup());
            consumer.start();
            log.info("{}cloudMq消费者初始化结束", RocketMqConstants.LOG_PREFIX);
        } catch (Exception e) {
            e.printStackTrace();
            throw new ServiceException(e);
        }
    }

    /**
     * 获取需要接收的tag
     *
     * @param topic
     * @return
     */
    private String getTags(String topic) {

        MqConstants.MQ_TOPIC mqTopic = MqConstants.MQ_TOPIC.getMqTopic(topic.trim().toUpperCase());
        if (StringUtils.isBlank(mqTopic.getTags()) || "*".equalsIgnoreCase(mqTopic.getTags())) {
            throw new ServiceException("禁止配置tag=null or tag=*的配置");
        }
        StringBuilder tagBuilder = new StringBuilder();
        String[] tags = mqTopic.getTags().split("\\|\\|");
        int i = 0;
        for (String tag : tags) {
            tagBuilder.append(tag);
            /*if (!mqTopic.getBroadCast()) {
                tagBuilder.append("&&").append(hosId);
            }*/
            if (++i < tags.length) {
                tagBuilder.append("||");
            }
        }
        return tagBuilder.toString();
    }

    /**
     * mq地址
     *
     * @return
     */
    protected abstract String getNameSrv();

    /**
     * 消费者group
     *
     * @return
     */
    protected abstract String getConsumerGroup();

    /**
     * topic
     *
     * @return
     */
    protected abstract String getTopic();

    /**
     * 消息者启用标志
     *
     * @return
     */
    protected abstract Boolean getConsumerEnable();
}